package sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 自定义功能丰富的Sink 继承RichSinkFunction
 */
public class CustomSinkDemo2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> data = env.socketTextStream("localhost", 8888);

        MySink mySink = new MySink();

        // 设置sink名称
        data.addSink(mySink).name("MyPrint");


        env.execute("Custom Sink");
    }

    public static class MySink extends RichSinkFunction<String> {

        private int number;

        // sink初始化调用 调用次数跟sink的并行度有关
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("invoke open.....");
            number = getRuntimeContext().getIndexOfThisSubtask() + 1;
        }
        // sink关闭时调用 调用次数和sink的并行度有关
        @Override
        public void close() throws Exception {
            System.out.println("invoke close......");
        }

        // 处理数据的方法 每一条数据就会调用一次
        @Override
        public void invoke(String value, Context context) throws Exception {
            System.out.println("invoke invoke......");
            System.out.println(number + "> " + value);
        }
    }
}
